"""
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    created by lei.chang on '24/05/2023'
    comment: 服务记录相关
"""
import asyncio
from datetime import datetime
from datetime import timedelta
import json

from project.core.exception import BizError
from project.core.base import BodyParams, Select
from project.core.function import serializer
from project.model.database.message import Message, MessageDevice
from project.micro_service.zeus import Zeus
from project.service.common.mqtt import MQTTService


class MessageService:

    @staticmethod
    async def do_message_push(session, body_params: BodyParams):
        """ 消息推送
        :param session:
        :param body_params:
        :return:
        """
        message = await Message.fetch_one(session, wheres=(Message.third_party_id == body_params.third_party_id, Message.third_party_msg_id == body_params.third_party_msg_id))
        if message:
            raise BizError('消息重复推送')

        # 如果expire_time为空，则默认为1天
        if not body_params.expire_time:
            body_params.expire_time = datetime.now() + timedelta(days=1)

        message = await Message.create_modify(session, **body_params.dict())
        if not message:
            raise BizError('消息推送失败')
        # 将 MQTT 推送逻辑放到后台处理
        asyncio.create_task(MessageService.push_message_to_devices(message, body_params.third_party_id))

        # 立即返回消息对象
        return message

    @staticmethod
    async def get_message_list(session, device):
        """ 查询指定设备的消息列表
        :param session: 数据库连接
        :param device: 设备信息
        :return:
        """
        # 获取第三方的消息列表
        organization = await Zeus.get_organization_detail(code=device.organization_code)
        if not organization:
            raise BizError('获取机构信息失败')

        sub_query = Select(selects=(MessageDevice.message_id), wheres=(MessageDevice.dev_id == device.dev_id)).subquery()
        messages = await Message.fetch_all(session,
                                           wheres=(Message.third_party_id == organization['third_party_id'], Message.id.notin_(sub_query), Message.expire_time > datetime.now()),
                                           orders=(Message.update_time.desc(),))

        return [v.dict() for v in messages]

    @staticmethod
    async def push_message_to_devices(message, third_party_id):
        """ 推送消息到设备
        :param message: 消息对象
        :param third_party_id: 第三方机构ID"
        """
        organization = await Zeus.get_organization_detail(third_party_id=third_party_id)
        if not organization:
            raise BizError('获取机构信息失败')

        devices = await Zeus.get_device_list(organization_id=organization['id'])
        if not devices:
            raise BizError('获取设备列表失败')

        push_data = serializer(message.dict())
        tasks = []
        for device in devices:
            tasks.append(
                MQTTService.push(
                    topic=f"message/{device['code']}",
                    message=json.dumps(push_data, ensure_ascii=False).encode('utf-8')
                )
            )
        await asyncio.gather(*tasks)  # 并发推送
